package com.zfei.firstflink;

import com.zfei.firstflink.model.XexHomeOrder;
import com.zfei.firstflink.utils.MysqlSource;
import com.zfei.firstflink.utils.MysqlSource2;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import scala.Tuple3;
import scala.Tuple4;

import java.math.BigDecimal;
import java.sql.Timestamp;

/***
 *$com.zfei.firstflink.TestWindow2
 *@author:jiwenfei
 *
 *@version:
 *
 *createTime:2018/8/21:上午10:20
 *
 *description:
 */
public class TestWindow2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStream<XexHomeOrder> orderStream=env.addSource(new MysqlSource2());
        orderStream.filter(item->item.getOrderType()==8).flatMap(new FlatMapFunction<XexHomeOrder, Tuple4<Integer,BigDecimal,Timestamp,Long>>() {
            @Override
            public void flatMap(XexHomeOrder item, Collector<Tuple4<Integer, BigDecimal, Timestamp,Long>> collector) throws Exception {
                collector.collect(new Tuple4<Integer,BigDecimal,Timestamp,Long>(item.getState(),item.getPrice(),item.getOrderTime(),1L));
            }
        }).keyBy(item->item._1()).window(TumblingEventTimeWindows.of(Time.days(30)))
                .reduce(new ReduceFunction<Tuple4<Integer, BigDecimal, Timestamp,Long>>() {
                    @Override
                    public Tuple4<Integer, BigDecimal, Timestamp, Long> reduce(Tuple4<Integer, BigDecimal, Timestamp, Long> reuslt, Tuple4<Integer, BigDecimal, Timestamp, Long> t1) throws Exception {
                        return new Tuple4<Integer, BigDecimal, Timestamp, Long> (t1._1(),t1._2().add(reuslt._2()),t1._3(),reuslt._4()+t1._4());
                    }
                })
//                .map(new MapFunction<Tuple4<Integer,BigDecimal,Timestamp,Long>, Tuple3<Integer,BigDecimal,Timestamp>>() {
//                    @Override
//                    public Tuple3<Integer, BigDecimal, Timestamp> map(Tuple4<Integer, BigDecimal, Timestamp, Long> t1) throws Exception {
//                        return new Tuple3<>(t1._1(),t1._2().divide( BigDecimal.valueOf(t1._4())),t1._3());
//                    }
//        })
        .print();
        env.execute("dd");


    }
}
